性能设计篇之"异步处理" [2026重制版]

# docker-compose-kafka.yml
version: '3.8'

services:
  # Kafka Broker 1
  kafka-1:
    image: apache/kafka:3.7.0
    container_name: kafka-1
    ports:
      - "9092:9092"
      - 9093:9093   # Controller
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-1:9093
      KAFKA_CONTROLLER_LISTENERS: CONTROLLER://:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_PLAINTEXT_LISTENERS: PLAINTEXT://:9092
      KAFKA_PLAINTEXT_HOST_LISTENERS: PLAINTEXT_HOST://:29092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-1:9092
      KAFKA_CLUSTER_ID: MkU3EABMTcCO5JxWsUYcRA
      KAFKA_LOG_DIRS: /var/lib/kafka/data
      KAFKA_NUM_PARTITIONS: 6
      KAFKA_DEFAULT_REPLICATION_FACTOR: 3
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      JMX_PORT: 9999
    volumes:
      - kafka-1-data:/var/lib/kafka/data
    networks:
      - kafka-net

  # Kafka UI (管理界面)
  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    container_name: kafka-ui
    ports:
      - "8080:8080"
    environment:
      DYNAMIC_CONFIG_ENABLED: "true"
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka-1:9092
    depends_on:
      - kafka-1
    networks:
      - kafka-net

volumes:
  kafka-1-data:

networks:
  kafka-net:
    driver: bridge

# application-kafka.yml
spring:
  kafka:
    bootstrap-servers: kafka-1:9092,kafka-2:9092,kafka-3:9092
    producer:
      # ====== 关键序列化器 ======
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      # ====== 可靠性配置 ======
      acks: all                    # 等待所有 ISR 副本确认
      retries: 3                   # 重试次数
      retry-backoff-ms: 100        # 重试退避
      max-in-flight-request-per-connection: 5
      enable-idempotence: true     # 幂等生产者（精确一次语义）
      compression-type: lz4       # 压缩算法
      batch-size: 16384            # 批次大小 (16KB)
      linger-ms: 5                 # 批次等待时间
      buffer-memory: 33554432      # 缓冲区大小 (32MB)
      # ====== 安全配置 ======
      properties:
        security.protocol: SASL_SSL
        sasl.mechanism: SCRAM-SHA-512
        sasl.jaas.config: |
          org.apache.kafka.common.security.scram.ScramLoginModule required
          username="${KAFKA_USERNAME}"
          password="${KAFKA_PASSWORD}";

    consumer:
      group-id: order-service-group
      auto-offset-reset: earliest
      enable-auto-commit: false       # 手动提交 offset
      auto-commit-interval: 5000
      max-poll-records: 500             # 每次 poll 最大条数
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "*"
        isolation.level: read_committed  # 事务隔离级别
        security.protocol: SASL_SSL
        sasl.mechanism: SCRAM-SHA-512
        sasl.jaas.config: |
          org.apache.kafka.common.security.scram.ScramLoginModule required
          username="${KAFKA_USERNAME}"
          password="${KAFKA_PASSWORD}";

    listener:
      type: BATCH                     # 批量消费
      concurrency: 3                   # 并发消费者数
      ack-mode: MANUAL_IMMEDIATE      # 手动即时确认
      poll-timeout: 3000               # poll 超时

@Service
@RequiredArgsConstructor
@Slf4j
public class OrderEventPublisher {

    private final KafkaTemplate<String, Object> kafkaTemplate;
    private final TransactionTemplate transactionTemplate;

    /**
     * 发送订单创建事件（带事务）
     */
    public void publishOrderCreated(OrderCreatedEvent event) {
        try {
            // 使用事务模板确保 DB 操作和消息发送原子性
            transactionTemplate.execute(status -> {
                // 1. 数据库操作已在业务层完成

                // 2. 发送 Kafka 消息
                SendResult<String, Object> result = kafkaTemplate.send(
                        "order-events",                          // Topic
                        event.getOrderId(),                      // Key（决定分区）
                        event                                    // Value
                    ).get(5, TimeUnit.SECONDS);              // 同步等待确认

                log.info("订单事件已发送: topic={}, partition={}, offset={}",
                    result.getRecordMetadata().topic(),
                    result.getRecordMetadata().partition(),
                    result.getRecordMetadata().offset());

                return null;
            });
        } catch (Exception e) {
            log.error("发送订单事件失败: orderId={}", event.getOrderId(), e);
            // 发送补偿：写入本地消息表或使用可靠消息方案
            saveToOutbox(event);
            throw new MessagePublishException("消息发送失败", e);
        }
    }

    /**
     * 发送到延迟 Topic（实现延迟消息）
     * 方案：使用不同 TTL 的 Topic 或 Pulsar 延迟消息
     */
    public void publishWithDelay(String topic, Object message, long delayMs) {
        kafkaTemplate.send(topic, message);
        // Kafka 原生不支持延迟消息，可通过以下方式实现：
        // 1. 使用时间轮桶（如 DelayQueue）
        // 2. 迁移到 Pulsar（原生支持）
        // 3. 使用 Kafka Streams 的 windowed 操作
    }
}

@Service
@RequiredArgsConstructor
@Slf4j
public class OrderEventConsumer {

    private final InventoryService inventoryService;
    private final PaymentService paymentService;
    private final KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * 消费订单创建事件 — 库存扣减
     */
    @KafkaListener(
        topics = "order-events",
        groupId = "inventory-consumer-group",
        containerFactory = "kafkaListenerContainerFactory"
    )
    public void handleOrderCreated(
            @Payload OrderCreatedEvent event,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            @Header(KafkaHeaders.OFFSET) long offset,
            Acknowledgment acknowledgment,
            Consumer<?, ?> consumer) {

        String messageId = event.getOrderId() + "-" + partition + "-" + offset;

        try {
            log.info("收到订单事件: orderId={}, partition={}, offset={}",
                event.getOrderId(), partition, offset);

            // 1. 幂等性检查（防止重复消费）
            if (inventoryService.isProcessed(messageId)) {
                log.info("消息已处理过，跳过: {}", messageId);
                acknowledgment.acknowledge();
                return;
            }

            // 2. 业务处理：扣减库存
            boolean success = inventoryService.deductStock(
                event.getProductId(),
                event.getQuantity(),
                event.getOrderId()
            );

            if (success) {
                // 3. 标记已处理（幂等保证）
                inventoryService.markAsProcessed(messageId);
                // 4. 手动确认
                acknowledgment.acknowledge();
                log.info("库存扣减成功: orderId={}, productId={}",
                    event.getOrderId(), event.getProductId());
            } else {
                // 库存不足，发送补偿事件
                kafkaTemplate.send(
                    "order-compensation-events",
                    new StockInsufficientEvent(event.getOrderId())
                );
                acknowledgment.acknowledge();
            }

        } catch (DuplicateMessageException e) {
            // 幂等异常，直接确认
            log.warn("幂等冲突: {}", e.getMessage());
            acknowledgment.acknowledge();

        } catch (Exception e) {
            log.error("处理订单事件失败: {}", messageId, e);
            // 不确认消息，触发重试
            // 注意：需要配置死信队列避免无限重试
        }
    }

    /**
     * 死信队列消费者 — 处理多次重试仍失败的消息
     */
    @KafkaListener(
        topics = "order-events.DLT",
        groupId = "dlt-handler-group"
    )
    public void handleDeadLetter(@Payload Object message) {
        log.error("进入死信队列的消息: {}", message);
        // 发送告警通知运维人员介入
        alertService.sendAlert("Kafka DLT", message.toString());
    }
}

/**
 * 领域事件基类
 */
public abstract class DomainEvent implements Serializable {
    private final String eventId;         // 全局唯一事件 ID
    private final String eventType;       // 事件类型
    private final String aggregateId;     // 聚合根 ID
    private final long timestamp;         // 事件时间戳
    private final String version;         // 事件版本
    private final Map<String, String> metadata;  // 元数据
}

/**
 * 具体事件示例：订单已创建
 */
@Data
@Builder
public class OrderCreatedEvent extends DomainEvent {
    private String orderId;
    private String userId;
    private List<OrderItem> items;
    private BigDecimal totalAmount;
    private String shippingAddress;

    /** 事件版本控制 */
    public static final String EVENT_TYPE = "order.created";
    public static final String VERSION = "2.0";
}

/**
 * 事件命名规范建议：
 *
 * 格式: {聚合}.{动作}PastTense
 * 示例:
 *   - order.created        (订单已创建)
 *   - order.paid           (订单已支付)
 *   - order.cancelled      (订单已取消)
 *   - inventory.reserved   (库存已预留)
 *   - inventory.released   (库存已释放)
 *   - payment.completed    (支付已完成)
 *   - user.registered      (用户已注册)
 */

@Service
@Transactional
public class OrderService {

    private final OrderMapper orderMapper;
    private final OutboxMapper outboxMapper;

    /**
     * 下单 — 使用本地消息表保证一致性
     */
    public OrderResult createOrder(CreateOrderCommand command) {
        // 1. 创建订单
        Order order = Order.create(command.getUserId(), command.getItems());
        orderMapper.insert(order);

        // 2. 写入本地消息表（同一事务！）
        OutboxMessage outbox = OutboxMessage.builder()
            .aggregateId(order.getId())
            .eventType(OrderCreatedEvent.EVENT_TYPE)
            .payload(JSON.toJSONString(
                OrderCreatedEvent.builder()
                    .orderId(order.getId())
                    .userId(command.getUserId())
                    .items(command.getItems())
                    .totalAmount(order.getTotalAmount())
                    .build()))
            .status(MessageStatus.PENDING)
            .createdTime(LocalDateTime.now())
            .build();
        outboxMapper.insert(outbox);

        // 3. 事务提交后，消息与订单原子性保存
        return OrderResult.success(order.getId());
    }
}

/**
 * 消息转发器 — 将本地消息表中的消息投递到 Kafka
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class OutboxRelayJob {

    private final OutboxMapper outboxMapper;
    private final KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * 定时扫描并发送消息（每秒执行）
     */
    @Scheduled(fixedDelay = 1000)
    public void relayPendingMessages() {
        // 查询待发送的消息（限制批次大小）
        List<OutboxMessage> pendingMessages = outboxMapper.selectPending(
            LocalDateTime.now().minusMinutes(5),  // 5分钟前的也重试
            200                                  // 每批最多200条
        );

        for (OutboxMessage msg : pendingMessages) {
            try {
                SendResult result = kafkaTemplate.send(
                    msg.getEventType(),
                    msg.getAggregateId(),
                    JSON.parseObject(msg.getPayload(), Object.class)
                ).get(3, TimeUnit.SECONDS);

                // 更新状态为已发送
                outboxMapper.markSent(msg.getId(),
                    result.getRecordMetadata().offset());

            } catch (Exception e) {
                log.error("消息投递失败: id={}, aggregateId={}",
                    msg.getId(), msg.getAggregateId(), e);
                // 增加重试计数
                outboxMapper.incrementRetryCount(msg.getId());
            }
        }
    }
}